package com.atguigu.tabletest

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.scala._
import org.apache.flink.table.api.{DataTypes, Table}
import org.apache.flink.table.descriptors._


/**
 *
 * @description: 更新聚合结果直接输出到es自动更新
 * @time: 2021-03-15 15:35
 * @author: baojinlong
 **/
object Example05EsOutputTest {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度
    environment.setParallelism(1)

    // 创建表的执行环境
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(environment)
    // 连接到kafka
    val filePath: String = "E:/big-data/FlinkTutorial/src/main/resources/sensor.data"

    tableEnv
      .connect(new FileSystem().path(filePath))
      .withFormat(new Csv())
      .withSchema(
        new Schema()
          .field("id", DataTypes.STRING)
          .field("timestamp", DataTypes.BIGINT)
          .field("temperature", DataTypes.DOUBLE))
      .createTemporaryTable("esInputTable")

    // 查询转换进行转换操作
    val sensorTable: Table = tableEnv.from("esInputTable")
    val resultTable: Table = sensorTable
      .select('id, 'temperature)
      .filter('id === "sensor_1")
    // 聚合转换
    val aggTable: Table = sensorTable
      .groupBy("id") // 基于id分组,后期es会根据这个id来更新, _id="sensor_1" _source: "id":"sensor_1" "count":5
      .select('id, 'id.count as 'count)


    // 输出到es
    tableEnv
      .connect(
        new Elasticsearch()
          .version("6")
          .host("localhost", 9092, "http")
          .index("sensor")
          .documentType("doc")
      )
      .inUpsertMode // 如果用了appendMode则就能输出resultTable
      .withFormat(new Json())
      .withSchema(
        new Schema()
          .field("id", DataTypes.STRING)
          .field("count", DataTypes.BIGINT)
      )
      .createTemporaryTable("esOutputTable")
    aggTable.insertInto("esOutputTable")

    environment.execute("es pipeline test")
  }

}
